package com.mysite.seckill.web.service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import com.alibaba.fastjson.JSON;
import com.mysite.seckill.commons.common.Constant;
import com.mysite.seckill.commons.exception.BusinessException;
import com.mysite.seckill.commons.po.SeckillActivity;
import com.mysite.seckill.commons.vo.DataTableResultVo;
import com.mysite.seckill.commons.vo.SeckillActivityVo;
import com.mysite.seckill.web.mapper.SeckillActivityMapper;
import com.mysite.seckill.web.util.UUIDUtil;

@Service
public class SeckillActivityService {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private DefaultRedisScript<Object> redisScript;

    @Autowired
    private SeckillActivityMapper seckillActivityMapper;

    public String create(SeckillActivity activity) {
        String sn = UUIDUtil.generate();
        activity.setSn(sn);
        if (activity.getStartTime() == null) {
            activity.setStartTime(new Date());
        }
        seckillActivityMapper.insertSeckillActivity(activity);
        String json = JSON.toJSONString(activity);
        redisTemplate.opsForHash().put(Constant.REDIS_SECKILL_ACTIVITY_MAP, sn, json);
        return sn;
    }

    @SuppressWarnings("unchecked")
    public boolean join(int userId, String sn, int seckillCount) {

        List<String> keys = new ArrayList<>();
        keys.add(Constant.REDIS_SECKILL_ACTIVITY_MAP);
        keys.add(sn + Constant.REDIS_SECKILL_USER_EXISTED_MAP_SUFFIX);
        Object[] args = new Object[] { sn, userId + "", seckillCount + "", System.currentTimeMillis() + "" };

        List<Object> list = (List<Object>) redisTemplate.execute(redisScript, keys, args);
        int retCode = ((Long) list.get(0)).intValue();
        if (retCode > 0) {
            throw new BusinessException(retCode);
        }

        kafkaTemplate.send(Constant.KAFKA_SECKILL_TOPIC, list.get(1).toString());

        /*ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(Constant.KAFKA_SECKILL_TOPIC,
                list.get(1).toString());
        if (future != null) {
            try {
                SendResult<String, String> sendResult = future.get();
                RecordMetadata metadata = sendResult.getRecordMetadata();
                System.err.println("partition: " + metadata.partition() + " ;; offset: " + metadata.offset());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }*/

        return true;
    }

    public DataTableResultVo getActivities(Map<String, Object> searchMap) {

        Integer count = seckillActivityMapper.selectActivityCount(searchMap);
        List<SeckillActivityVo> list = null;
        if (count > 0) {
            list = seckillActivityMapper.selectActivities(searchMap);
        }
        return new DataTableResultVo(Integer.parseInt(searchMap.get("draw").toString()), count, list);
    }
}
